package com.zhangxiaofanfan.learn.consumer;

import com.zhangxiaofanfan.learn.utils.KafkaUtils;
import lombok.AllArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.locks.LockSupport;

/**
 * Kafka 学习过程中的 消费者
 *
 * @author zhangxiaofanfan
 * @date 2023-10-17 14:45:32
 */
public class ConsumerFastStart {
    public static final String groupId = "first-group";
    public static final String quitStr = "quit";
    public static void main(String[] args) throws InterruptedException {
        Properties properties = KafkaUtils.getKafkaConsumerProperties();
        // 设置消费组的名称
        properties.put("group.id", groupId);
        // 设置一个消费者客户端实例
        KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(properties);
        KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(properties);
        KafkaConsumer<String, String> consumer3 = new KafkaConsumer<>(properties);
        // 订阅主题
        consumer1.subscribe(Collections.singletonList(KafkaUtils.topic));
        consumer2.subscribe(Collections.singletonList(KafkaUtils.topic));
        consumer3.subscribe(Collections.singletonList(KafkaUtils.topic));
        // 循环消费消息
        Thread consumer1Thread = new Thread(new Consumer(consumer1));
        Thread consumer2Thread = new Thread(new Consumer(consumer2));
        Thread consumer3Thread = new Thread(new Consumer(consumer3));

        consumer1Thread.start();
        consumer2Thread.start();
        consumer3Thread.start();

        LockSupport.park();
    }

    @AllArgsConstructor
    public static class Consumer implements Runnable {
        KafkaConsumer<String, String> consumer;

        @Override
        public void run() {
            boolean isQuit = false;
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    if (quitStr.equals(record.value())) {
                        System.out.println(Thread.currentThread().getName() + ": " + "消费者退出");
                        consumer.close();
                        return ;
                    } else {
                        System.out.println(Thread.currentThread().getName() + ": " + record);
                    }
                }
            }
        }
    }
}
